/****

    activequant - activestocks.eu

    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
    the Free Software Foundation; either version 2 of the License, or
    (at your option) any later version.

    This program is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    GNU General Public License for more details.

    You should have received a copy of the GNU General Public License along
    with this program; if not, write to the Free Software Foundation, Inc.,
    51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.

	
	contact  : contact@activestocks.eu
    homepage : http://www.activestocks.eu

****/
package org.activequant.data.util.subscription;

import org.activequant.core.domainmodel.InstrumentSpecification;
import org.activequant.core.domainmodel.data.MarketDataEntity;
import org.activequant.core.types.TimeFrame;
import org.activequant.data.retrieval.ISubscription;
import org.activequant.data.retrieval.SubscriptionSourceBase;
import org.activequant.data.retrieval.filtering.IDataFilter;
import org.activequant.util.exceptions.SubscriptionException;
import org.activequant.util.pattern.events.IEventListener;
import org.apache.log4j.Logger;

/**
 * Generic filter of live quotes. Uses pluggable strategy to implement 
 * filtering decisions.<br/>
 * An abstract class. SubscriptionSourceFilterBase&lt;T extends MarketDataEntity&lt;T&gt;&gt; extends SubscriptionSourceBase&lt;T&gt;.<br/>
 * Defines a nested DataSubscription class. Holds the following associated variables:
 * <ul>
 * <li>filter(IDataFilter&lt;T&gt;)</li>
 * </ul>
 * Holds the following inherited variables:
 * <ul>
 * <li>activeBackends(Map&lt;Topic, Subscription&gt;)</li>
 * <li>activeSubscriptions(Queue&lt;SubscriptionDelegate&gt;)</li>
 * </ul>
 *       <b>History:</b><br/>
 * - [04.12.2007] Created (Mike Kroutikov)<br>
 * 
 * @author Mike Kroutikov
 */
public abstract class SubscriptionSourceFilterBase<T extends MarketDataEntity<T>> extends SubscriptionSourceBase<T> {
	
	protected final Logger log = Logger.getLogger(getClass());
	/**
	 * private IDataFilter&lt;T&gt; filter = new IDataFilter&lt;T&gt;() {...}<br/>
	 * A filter whose <code>boolean evaluate(T data)</code> method always returns true
	 */
	private IDataFilter<T> filter = new IDataFilter<T>() {
		public boolean evaluate(T data) {
			return true;
		}
	};	
	/**
	 * An empty SubscriptionSourceFilterBase&lt;T extends MarketDataEntity&lt;T&gt;&gt; (extends SubscriptionSourceBase&lt;T&gt;) constructor
	 */
	public SubscriptionSourceFilterBase() {
		super();
	}
	/**
	 * returns the associated filter(IDataFilter&lt;T&gt;)
	 * @return
	 */
	public IDataFilter<T> getFilter() {
		return filter;
	}
	/**
	 * sets the associated filter(IDataFilter&lt;T&gt;) with the given val(IDataFilter&lt;T&gt;)
	 * @param val
	 */
	public void setDataFilter(IDataFilter<T> val) {
		filter = val;
	}
	/**
	 * A nested class. DataSubscription extends Subscription. Holds the following associated variables:
	 * <ul>
	 * <li>subscription(ISubscription&lt;T&gt;)</li>
	 * <li>listener(IEventListener&lt;T&gt;)</li>
	 * </ul>
	 * @author Dimitar
	 *
	 */
	class DataSubscription extends Subscription {
		/**
		 * private final ISubscription&lt;T&gt; subscription;
		 */
		private final ISubscription<T> subscription;
		/**
		 * constructs a DataSubscription(extends Subscription) using the given s(ISubscription&lt;T&gt;) to set its associated subscription(ISubscription&lt;T&gt;)
		 * @param s
		 */
		DataSubscription(ISubscription<T> s) {
			subscription = s;
		}
		/**
		 * private final IEventListener&lt;T&gt; listener = new IEventListener&lt;T&gt;() {...}<br/>
		 * A listener which uses the filter(IDataFilter&lt;T&gt;) to evaluate which given events(T) will get propagated and which ones won't
		 */
		private final IEventListener<T> listener = new IEventListener<T>() {
			public void eventFired(T event) {
				if(filter.evaluate(event)) {
					fireEvent(event);
				}
			}
		};
		/**
		 * registers the associated listener(IEventListener&lt;T&gt;) to the associated subscription(ISubscription&lt;T&gt;) and then activates that subscription
		 */
		@Override
		protected void handleActivate() {
			subscription.addEventListener(listener);
			subscription.activate();
		}
		/**
		 * stops the associated listener(IEventListener&lt;T&gt;) from listening to the associated subscription(ISubscription&lt;T&gt;) and cancels the subscription
		 */
		@Override
		protected void handleCancel() {
			subscription.removeEventListener(listener);
			subscription.cancel();
		}
	}
	/**
	 * An abstract method
	 * @param spec
	 * @param timeFrame
	 * @return
	 * @throws Exception
	 */
	protected abstract ISubscription<T> openSubscription(InstrumentSpecification spec, TimeFrame timeFrame) throws Exception;
	/**
	 * creates and returns a DataSubscription (extends Subscription) for the given spec(InstrumentSpecification) and timeFrame(TimeFrame).<br/>
	 * A DataSubscription uses the associated filter(IDataFilter&lt;T&gt;) to evaluate which events which be propagated and which ones will be filtered out.
	 */
	@Override
	protected DataSubscription createSubscription(InstrumentSpecification spec, TimeFrame timeFrame) {
		try {
			return new DataSubscription(openSubscription(spec, timeFrame));
		} catch(Exception ex) {
			throw new SubscriptionException(ex);
		}
	}
}